package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.DimUtil;
import com.atguigu.utils.DruidDSUtil;
import com.atguigu.utils.PhoenixUtil;
import org.apache.flink.api.common.functions.RichFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;

/**
 * @className: DimSinkFunction
 * @author: LinCong
 * @description:
 * @date: 2023/1/13 22:46
 * @version: 1.0
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private DruidDataSource druidDataSource = null;

    @Override
    public void open(Configuration parameters) throws Exception {
//        一个并行度一个连接池
        druidDataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
//        获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");

//        跟redis对接，获取数据类型
        String type = value.getString("type");
//        如果为更新数据类型，则需要删除redis中的数据
        if ("update".equals(type)){
            //        phonix中表明会自动转换为大写（如果不加双引号），为了和redis 表明统一用大写
            DimUtil.delDimInfo(sinkTable.toUpperCase(),data.getString("id"));
        }

//        写出数据
        //{"sinkTable":"dim_xxx","database":"gmall","table":"base_trademark","type":"insert","ts":1673323672,"data":{"id":1,"tm_name":"三星"}}
        try {
            PhoenixUtil.upsertValues(connection,sinkTable,data);
        } catch (SQLException throwables) {
            System.out.println("维度数据写入异常");
            throwables.printStackTrace();
        }

//        归还连接
        connection.close();
    }
}
